#include <iostream>
#include <cstdio>
#include "asio.hpp"

using asio::ip::tcp;

class Session : public std::enable_shared_from_this<Session>
{
public:
    enum
    {
        E_SESSION_BUFFER_LEN = 16,
        E_MAX_PING_PONG_TIMES = 20,
        E_SEND_DATA_INTERVAL_SECONDS = 10,
    };
public:
    Session(asio::io_context& io, tcp::socket socket)
        : send_timer_(io),
          socket_(std::move(socket))
    {
        /* fprintf(stdout, "session established, %s:%d<-->%s:%d\n",
                 socket_.local_endpoint().address().to_string().c_str(),
                 socket_.local_endpoint().port(),
                 socket_.remote_endpoint().address().to_string().c_str(),
                 socket_.remote_endpoint().port());*/
        ++session_count;
    }
    ~Session()
    {
        /*fprintf(stdout, "session closed, %s:%d<-->%s:%d\n",
                socket_.local_endpoint().address().to_string().c_str(),
                socket_.local_endpoint().port(),
                socket_.remote_endpoint().address().to_string().c_str(),
                socket_.remote_endpoint().port());*/
        --session_count;
    }
    void Start()
    {
        DoWrite(shared_from_this());
    }
private:
    void DoRead()
    {
        auto self(shared_from_this());
        socket_.async_read_some(asio::buffer(buffer_, E_SESSION_BUFFER_LEN),
                                [this, self](std::error_code ec, std::size_t length)
        {
            if(ec)
            {
                fprintf(stdout, "DoRead Error: %s\n", ec.message().c_str());
            }
            else
            {
                ++ping_pong_times_;
                if(ping_pong_times_ < E_MAX_PING_PONG_TIMES)
                {
                    send_timer_.expires_after(asio::chrono::seconds(E_SEND_DATA_INTERVAL_SECONDS));
                    send_timer_.async_wait(std::bind(&Session::DoWrite, this, self));
                }
            }
        });
    }
    void DoWrite(std::shared_ptr<Session> self)
    {
        asio::async_write(socket_, asio::buffer(buffer_, E_SESSION_BUFFER_LEN),
                          [this, self](std::error_code ec, std::size_t /*length*/)
        {
            if(ec)
            {
                fprintf(stdout, "DoWrite Error: %s\n", ec.message().c_str());
            }
            else
            {
                DoRead();
            }
        });
    }
public:
    static std::size_t session_count;
private:
    std::array<char, E_SESSION_BUFFER_LEN> buffer_;
    tcp::socket socket_;
    std::size_t ping_pong_times_ = 0;
    asio::steady_timer send_timer_;
};
std::size_t Session::session_count = 0;

class ClientCluster
{
public:
    enum
    {
        E_MAX_SESSION_COUNT = 10000,
        E_CONNECT_INTERVAL_MILLI_S = 1,
    };
public:
    ClientCluster(asio::io_context& io, const char *host, const char *port)
        : io_context_(io),
          resolver_(io),
          connect_timer_(io),
          show_info_timer_(io)
    {
        endpoints_ = resolver_.resolve(host, port);
        show_info_timer_.expires_after(asio::chrono::seconds(1));
        show_info_timer_.async_wait(std::bind(&ClientCluster::ShowInfo, this));
    }
public:
    void Start()
    {
        connect_timer_.expires_after(asio::chrono::microseconds(E_CONNECT_INTERVAL_MILLI_S));
        connect_timer_.async_wait(std::bind(&ClientCluster::NewConnect, this, std::placeholders::_1));
    }
private:
    void NewConnect(const asio::error_code& ec)
    {
        if(ec)
        {
            fprintf(stdout, "NewConnect Error: %s\n", ec.message().c_str());
            return;
        }
        if(Session::session_count < E_MAX_SESSION_COUNT)
        {
            std::shared_ptr<tcp::socket> new_socket = std::make_shared<tcp::socket>(io_context_);
            asio::async_connect(*new_socket.get(), endpoints_,
                                [this, new_socket](const asio::error_code & ec, const tcp::endpoint & endpoint)
            {
                if(ec)
                {
                    fprintf(stdout, "async_connect Error: %s\n", ec.message().c_str());
                }
                else
                {
                    std::make_shared<Session>(io_context_, std::move(*new_socket.get()))->Start();
                }
            });
        }
        connect_timer_.expires_after(asio::chrono::microseconds(E_CONNECT_INTERVAL_MILLI_S));
        connect_timer_.async_wait(std::bind(&ClientCluster::NewConnect, this, std::placeholders::_1));
    }
    void ShowInfo()
    {
        fprintf(stdout, "session_count = %llu\n\r\b", Session::session_count);
        show_info_timer_.expires_after(asio::chrono::seconds(1));
        show_info_timer_.async_wait(std::bind(&ClientCluster::ShowInfo, this));
    }
private:
    asio::io_context & io_context_;
    tcp::resolver resolver_;
    tcp::resolver::results_type endpoints_;
    asio::steady_timer connect_timer_;
    asio::steady_timer show_info_timer_;
};

int main(int argc, char* argv[])
{
    const char * host = "127.0.0.1";
    const char * port = "8888";
    try
    {
        if(argc > 2) { host = argv[1]; }
        if(argc > 3) { port = argv[2]; }

        asio::io_context io_context(1);
        ClientCluster client_cluster(io_context, host, port);
        client_cluster.Start();
        io_context.run();
    }
    catch(std::exception& e)
    {
        std::cerr << "Exception: " << e.what() << "\n";
    }
    return 0;
}